/*
 * Copyright (c) 2007 Compl Yue Still, All Rights Reserved.
 */

/**
 * 
 */
package hbi.wire.java;

import hbi.wire.java.host.HostEnd;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;

public class WireTrafficHandler extends Thread
{

    private final Selector selector;

    public WireTrafficHandler() throws IOException
    {
        super("HBI Socket Wire Traffic Handler ");
        super.setDaemon(true);

        this.selector = Selector.open();
    }

    public int getNumChannels()
    {
        return selector.keys().size();
    }

    private static class C2A
    {
        final SocketChannel ch;

        final HostEnd hostEnd;

        C2A(SocketChannel ch, HostEnd hostEnd)
        {
            this.ch = ch;
            this.hostEnd = hostEnd;
        }
    }

    final Queue<C2A> c2as = new ConcurrentLinkedQueue<C2A>();

    public void handleChannel(SocketChannel channel, HostEnd hostEnd)
    {
        c2as.offer(new C2A(channel, hostEnd));
        selector.wakeup();
        if (!this.isAlive()) this.start();
    }

    private static class WireToken
    {
        final HostEnd hostEnd;

        ByteBuffer outgoing = null;

        WireToken(HostEnd hostEnd)
        {
            this.hostEnd = hostEnd;
        }
    }

    public void run()
    {
        try
        {
            while (!Thread.interrupted())
            {
                for (C2A c2a = c2as.poll(); c2a != null; c2a = c2as.poll())
                {
                    HostEnd hostEnd = c2a.hostEnd;
                    SelectionKey selKey = c2a.ch.register(selector, //
                        SelectionKey.OP_READ | SelectionKey.OP_WRITE, //
                        new WireToken(hostEnd));
                    hostEnd.gateway.setSelectionKey(selKey);
                }

                int readyChannels = selector.select();

                if (readyChannels > 0)
                {
                    for (Iterator<SelectionKey> kIt = selector.selectedKeys()
                        .iterator(); kIt.hasNext();)
                    {
                        SelectionKey selKey = kIt.next();
                        kIt.remove();
                        if (!selKey.isValid()) continue;

                        try
                        {
                            WireToken tk = (WireToken) selKey.attachment();
                            HostEnd hostEnd = tk.hostEnd;

                            if (selKey.isReadable())
                            {
                                hostEnd.readChannel((SocketChannel) selKey
                                    .channel());
                            }

                            if (selKey.isWritable())
                            {
                                if (tk.outgoing == null)
                                    tk.outgoing = hostEnd.gateway
                                        .pollOutgoingData();
                                if (tk.outgoing != null)
                                {
                                    ((SocketChannel) selKey.channel())
                                        .write(tk.outgoing);
                                    if (!tk.outgoing.hasRemaining())
                                        tk.outgoing = hostEnd.gateway
                                            .pollOutgoingData();
                                }
                            }
                        }
                        catch (Exception e)
                        {
                            e.printStackTrace();
                            selKey.channel().close();
                        }
                    }
                }

                for (SelectionKey selKey : selector.keys())
                {
                    if (!selKey.isValid()) continue;

                    try
                    {
                        WireToken tk = (WireToken) selKey.attachment();
                        HostEnd hostEnd = tk.hostEnd;
                        if (tk.outgoing == null)
                            tk.outgoing = hostEnd.gateway.pollOutgoingData();
                        int ops = 0;
                        if (hostEnd.isClosedByPeer())
                        {
                            if (tk.outgoing != null)
                                ops = SelectionKey.OP_WRITE;
                            else
                                selKey.channel().close();
                        }
                        else
                        {
                            if (tk.outgoing != null)
                                ops = SelectionKey.OP_WRITE
                                    | SelectionKey.OP_READ;
                            else
                                ops = SelectionKey.OP_READ;
                        }
                        if (selKey.isValid()) selKey.interestOps(ops);
                    }
                    catch (Exception e)
                    {
                        e.printStackTrace();
                        selKey.channel().close();
                    }
                }
            }
        }
        catch (IOException e)
        {
            e.printStackTrace();
        }
        finally
        {
            for (SelectionKey selKey : selector.keys())
            {
                try
                {
                    if (selKey.channel().isOpen()) selKey.channel().close();
                }
                catch (IOException e)
                {
                    e.printStackTrace();
                }
            }
            try
            {
                selector.close();
            }
            catch (IOException e)
            {
                e.printStackTrace();
            }
        }
    }

}
